package me.xhy.java.springboot3.webflux.projectreactor.s2java9flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * @author xhy
 * @since 2024-10-30 上午8:29
 */
public class S2Processor {

  public static void main(String[] args) throws InterruptedException {
    b();

    // 订阅默认使用 ForkJoinPool 线程池，需要保持主线程存在
    TimeUnit.SECONDS.sleep(100);
  }

  // Processor 类。 继承 SubmissionPublisher 是为了方便发送数据；实现 Processor ，
  // Processor 既是 Publisher 又是 Subscriber : Publisher | ProcessorA -> ProcessorB -> ProcessorC | Subscriber
  // onSubscribe、onNext、onError、onComplete 是 Flow.Processor 接口中 Subscriber 的方法，需要实现。
  // subscribe 是 Flow.Processor 接口中 Publisher 的方法，由父类 SubmissionPublisher 实现了。
  static class CertainProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

    // 绑定关系
    Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
      // 与订阅者写法一致
      System.out.println("Processor onSubscribe: " + Thread.currentThread());
      // 保存绑定关系
      this.subscription = subscription;
      // 找上游要一个数据
      this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
      // 数据到达
      System.out.println("Processor onNext: " + Thread.currentThread() + ", item: " + item);

      // 对数据进行加工
      item += " | *";

      // 使用 Publisher 的 submit() 发布数据 给下游
      submit(item);

      // 再请求一个数据
      subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
  }

  /*
  串联 Publisher Processor... Subscriber
  Publisher.subscribe(Processor)
  Processor.subscribe(Processor)...
  Processor.subscribe(Subscriber)
   */
  public static void b() throws InterruptedException {
    // 1. 一个简单直观的Publisher
    SubmissionPublisher publisher = new SubmissionPublisher();

    // 2.Processor
    CertainProcessor logicAProcessor = new CertainProcessor();
    CertainProcessor logicBProcessor = new CertainProcessor();


    // 3. Subscriber
    Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

      // 订阅关系
      Flow.Subscription subscription;

      // onXxx() 方法，在xxx事件发生时，执行这个回调

      // 在订阅时回调，会将订阅关系传递过来
      @Override
      public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Subscriber onSubscribe: " + Thread.currentThread());

        // 保存订阅关系
        this.subscription = subscription;
        // 在订阅时，request 数据，触发订阅者的 onNext
        this.subscription.request(1);
      }

      // 在下一个元素到达时=接收到数据
      @Override
      public void onNext(String item) {
        System.out.println("Subscriber onNext: " + Thread.currentThread() + ", item: " + item);

        this.subscription.request(1);

        if (item.equals("publish:10")) {
          // 取消订阅关系
          this.subscription.cancel();
        }
      }

      // 在错误发生时
      @Override
      public void onError(Throwable throwable) {
        System.out.println("onError: " + Thread.currentThread() + ", throwable: " + throwable);
      }

      @Override
      public void onComplete() {
        System.out.println("onComplete: " + Thread.currentThread());
      }
    };

    // 4. 绑定发布者和订阅者
    publisher.subscribe(logicAProcessor);
    logicAProcessor.subscribe(logicBProcessor);
    logicBProcessor.subscribe(subscriber);

    // 5. 发送数据
    // 先完成订阅，再发送数据。订阅关系创建完成前的事件，将会丢失
    for (int i = 0; i < 10; i++) {
      // submit()发布数据
      // SubmissionPublisher 内部的数组对象，缓存着submit()的数据。并且在存入数组之前，有锁控制 抑制并发的发生。
      publisher.submit("publish:" + i);
      System.out.println("发布数据的线程: " + Thread.currentThread());
    }

    // 不要马上报错
    TimeUnit.SECONDS.sleep(3);
    // closeExceptionally 会插队，没有接收到的 事件（消息事件、完成事件） 也不回到达 Subscriber 了
    publisher.closeExceptionally(new Throwable("发生错误"));

    // 错误事件发生后，继续发数据，将抛出异常
    TimeUnit.SECONDS.sleep(3);
    for (int i = 10; i < 20; i++) {
      // submit()发布数据
      // SubmissionPublisher 内部的数组对象，缓存着submit()的数据。并且在存入数组之前，有锁控制 抑制并发的发生。
      publisher.submit("publish:" + i);
      System.out.println("发布数据的线程: " + Thread.currentThread());
    }

    // 关闭发布者
    publisher.close();

  }


}
